package com.ruoyi.flink.chapten5;

/**
 * Copyright (c) 2020-2030 尚硅谷 All Rights Reserved
 * <p>
 * Project:  FlinkTutorial
 * <p>
 * Created by  wushengran
 */

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Trans1MapTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        // 进行转换计算,提取user字段
        // 1. 使用自定义类,实现mapFunction 接口   
        SingleOutputStreamOperator<String> result1 =  stream.map(new UserExtractor());
        result1.print();
        
        
        // 2使用匿名类实现MapFunction
        SingleOutputStreamOperator<String> result2 = stream.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event e) throws Exception {
                return e.user;
            }
        });
        result2.print();

        
        // 3.传入lambda表达式(data 为传入参数, data.user为返回参数)
        SingleOutputStreamOperator<String> result3 = stream.map(data -> data.user); 
        result3.print();
        

        env.execute();
    }
    
    // 自定义MapFunction
    public static class UserExtractor implements MapFunction<Event, String> {
        @Override
        public String map(Event e) throws Exception {
            return e.user;
        }
    }
}

